package com.xhg.mqtt.job;

import com.xhg.mqtt.constant.MqttConstants;
import com.xhg.mqtt.constant.MsgConstants;
import com.xhg.mqtt.core.MqttSimpleClient;
import com.xhg.mqtt.dto.EventCodeEnum;
import com.xhg.mqtt.listener.MessageEvent;
import com.xhg.mqtt.listener.MqttLoggerFactory;
import com.xhg.mqtt.util.PropertiesUtil;
import java.util.Map;

/**
 * 消息重发处理
 *
 * @author E.T
 * @version 1.0
 * @Description
 * @date 2018-07-27 19:52
 * @updateby
 * @updatedate
 * @since 1.0
 */
public class MessageTimer implements Runnable {
    private Map<String, MessageEvent> map;
    private Map<String, MessageEvent> failMsgMap;

    public MessageTimer(Map<String, MessageEvent> map, Map<String, MessageEvent> failMsgMap) {
        this.map = map;
        this.failMsgMap = failMsgMap;
    }

    @Override
    public synchronized  void run() {
        Thread.currentThread().setName("消息重发线程");
        //获得当前时间，得到的是毫秒数
        long start = System.currentTimeMillis();
        if (map.isEmpty()) {
            //退出定时器
            return;
        }
        try {
            for (String key : map.keySet()) {
                MessageEvent event = map.get(key);
                if (event == null) {
                   continue;
                }
                byte[] message = event.getMessage();
                String topic = event.getTopic();
                String eventCode = event.getEventCode();
                long end = event.getTimes();
                int count = event.getCount();
                if (MqttSimpleClient.getInstance() != null && MqttSimpleClient.getInstance().isConnect()) {
                    //10秒后还在队列里面的就重发
                    if (start - end > PropertiesUtil.RESEND_TIMES*1000) {
                        MessageEvent event2 = map.get(key);
                        if (event2 == null) {
                            continue;
                        }
                        //重发当条信息
                        MqttLoggerFactory.info("重发第" + count + "次" + "  消息Id:" + key + " 重发到topic->" + topic + "  指令码: " + eventCode + " 指令信息: " + EventCodeEnum.enumOf(eventCode).getDesc());
                        MqttSimpleClient.getInstance().publish(topic,MqttConstants.QOS, message);
                        //更新重发次数
                        event.setCount(event.getCount() + 1);
                        map.put(key, event);
                    }
                    if (event.getCount() >= PropertiesUtil.RESEND_COUNT) {
//                        MqttLoggerFactory.warn(MsgConstants.ATTEMPT_MANY_ATTEMPT);
                        //放入失败队列 进行入库操作
//                        failMsgMap.put(key, event);
                        //移除原来队列消息
                        map.remove(key);
                        //通知失败消息处理
//                        MessageRepeatQueue.getInstance().listenerFailMsg();
                    }
                }
            }
        } catch (Exception e) {
            MqttLoggerFactory.error(MsgConstants.RESEND_MESSAGE_EXCEPTION+e);
        }
    }
}
